package cn.com.sparkle.paxos.client;

import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition; 
import java.util.concurrent.locks.ReentrantLock;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.client.ConnectConfig.ConnectEvent;
import cn.com.sparkle.paxos.deamon.ReConnectThread;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.net.CallBack;
import cn.com.sparkle.paxos.net.NetNode;
import cn.com.sparkle.paxos.net.factory.NetClient;
import cn.com.sparkle.paxos.net.factory.NetFactory;

public class CommandProcessor implements Runnable {
	private final static Logger logger = Logger
			.getLogger(CommandProcessor.class);
	// private final static GetNewsSenatorsRequest GET_NEWS_SENATORS_REQUEST =
	// new GetNewsSenatorsRequest();
	
	private static NetClient nioSocketClient;
	private static ClientHandler handler;
	private static ReConnectThread reConnectThread = new ReConnectThread();
	
	
	private volatile String[] senator = {};
	private volatile NetNode node;
	private volatile ConnectConfig firstPriorConfig;
	private volatile HashMap<String, NetNode> activeNode = new HashMap<String, NetNode>();

	private ArrayBlockingQueue<Command> commandQueue;
//	private LinkedBlockingQueue<Command> runningCommand;
	
	private Entry root = new Entry();
	private AtomicInteger runningSize = new AtomicInteger(0);
	
//	private HashMap<Long, WatchCommand> runningWatchCommand = new HashMap<Long, WatchCommand>(); 
	
	private final int maxRunningSize;
	// private final int maxWaitRunSize;

	private ReentrantLock lock = new ReentrantLock();
	private Condition wakeupCondition = lock.newCondition();

	private ReentrantLock nodeChangeLock = new ReentrantLock();

	private volatile boolean isWakeup = false;

	private boolean isPromiseMaster;
	
	private static class Entry{
		private Command c;
		private Entry next;
		private Entry prev;
		
	}
	private static class CommandCallback implements CallBack {
		private Command c;
		private CommandProcessor processor;
		private Entry e;

		public CommandCallback(Command c, CommandProcessor processor,Entry e) {
			this.c = c;
			this.processor = processor;
			this.e = e;
		}

		@Override
		public void call(NetNode nnode, MessagePackage value) {
			Entry root = processor.getRoot();
			if(value.getIsLast()){ // if this is finished ,remove from unfinish list
				if(e.prev == root ){
					synchronized (root) {
						if(e.prev == root){
							root.next = e.next;
							if(e.next != null){
								e.next.prev = root;
							}
						}
					}
				}
				if(e.prev != root){
					e.prev.next = e.next;
					if(e.next != null){
						e.next.prev = e.prev;
					}
				}
				processor.getRunningSize().decrementAndGet();
				processor.wakeup();
			}
			c.finish(nnode, value);
		}
		@Override
		public void fail(NetNode nnode) {
			processor.wakeup();//lost the node
		}

	}
	
	// private CallBack callback = new CallBack() {
	//
	// @Override
	// public void call(NetNode nnode, MessagePackage value) {
	//
	// finish(nnode,value);
	// }
	//
	// @Override
	// public void fail(NetNode nnode) {
	// }
	// };

	private ConnectEvent connectEvent = new ConnectEvent() {
		@Override
		public void disconnect(String address, NetNode node) {
			synchronized (this) {
				HashMap<String, NetNode> newSet = new HashMap<String, NetNode>();
				newSet.putAll(activeNode);
				newSet.remove(address);
				activeNode = newSet;
				node.close();
			}
		}

		@Override
		public void connect(String address, NetNode node) {
			synchronized (this) {
				HashMap<String, NetNode> newSet = new HashMap<String, NetNode>();
				newSet.putAll(activeNode);
				newSet.put(address, node);
				activeNode = newSet;
			}
		}
	};

	public void runDeamon() {
		Thread t = new Thread(this);
		t.setName("CommandProcessor");
		t.start();
	}

	public static void init(String confPath,String netLayerType,EventsManager eventsManager) throws Throwable {
		nioSocketClient = NetFactory.makeClient(netLayerType);
		handler = new ClientHandler(nioSocketClient, reConnectThread,eventsManager);
		nioSocketClient.init(confPath, handler);
		reConnectThread.startThread();
	}

	public CommandProcessor(String[] senator, int maxRunningSize,
			int maxWaitRunSize, boolean isPromiseMaster) throws Throwable {
		this.isPromiseMaster = isPromiseMaster;
		this.senator = senator;
		this.maxRunningSize = maxRunningSize;
//		runningCommand = new LinkedBlockingQueue<Command>(maxRunningSize);
		commandQueue = new ArrayBlockingQueue<Command>(maxWaitRunSize);

		// for (String aa : senator) {
		String[] tmp = senator[0].split(":");
		firstPriorConfig = new ConnectConfig(senator[0], true, isPromiseMaster,
				connectEvent);
		boolean isSuccess = nioSocketClient.connect(tmp[0],Integer.parseInt(tmp[1]),
				firstPriorConfig).get();
		if (isSuccess) {
			firstPriorConfig.waitOnOpenEvent();
		}
		// }
	}

	public void changeSenator(String[] senator) throws Throwable {
		try {
			nodeChangeLock.lock();
			if (!senator[0].equals(this.senator[0])) {
				firstPriorConfig.setAutoReConnect(false);
				firstPriorConfig = new ConnectConfig(senator[0], true,
						isPromiseMaster, connectEvent);
				String[] tmp = senator[0].split(":");
				boolean isSuccess = nioSocketClient.connect(tmp[0], Integer.parseInt(tmp[1]), firstPriorConfig).get();
				if (isSuccess) {
					firstPriorConfig.waitOnOpenEvent();
				}
			}

			NetNode n = node;
			if (n != null) {
				n.close();
			}
			this.senator = senator;
		} finally {
			nodeChangeLock.unlock();
		}
	}

	public void wakeup() {
		try {
			lock.lock();
			isWakeup = true;
			wakeupCondition.signal();
		} finally {
			lock.unlock();
		}
	}

	int testCount = 0;
	long ct;

	public void run() {
		try {
			while (true) {
				//reactor
				try {
					lock.lock();
					if (!isWakeup) {
						wakeupCondition.await(); 
					}
					isWakeup = false;
				} finally {
					lock.unlock();
				}
				if(node != null && node.isClose()){
					node = null;
				}
				while (node == null) {
					// reconnect master
					reConnect();
					// redo
					Entry next = root;
					while((next = next.next) != null){
						if(!node.write(next.c.getRequest(), new CommandCallback(next.c, this, next))){
							logger.error("master disconnected!");
							node = null;
							Thread.sleep(1000);
							break;
						}
					}
				}

				while (commandQueue.peek() != null
						&& runningSize.get() != maxRunningSize) {
					if (testCount == 0) {
						ct = System.currentTimeMillis();
					}
					if (++testCount % 1000 == 0) {
						logger.debug("send " + testCount);
					}
					Command c = commandQueue.poll();
					Entry e = new Entry();
					e.c = c;
					synchronized (root) {
						e.next = root.next;
						if(root.next != null){
							root.next.prev = e;
						}
						root.next = e;
						e.prev = root;
					}
					runningSize.incrementAndGet();
					
					if (!node.write(c.getRequest(),
							new CommandCallback(c, this,e))) {
						logger.error("master disconnect!");
						node = null;
						wakeup();
						break;
					}
				}
				// logger.debug("runningCommand size:" + runningCommand.size());
			}

		} catch (InterruptedException e) {
			throw new RuntimeException("not support interruptedException", e);
		}
	};

	// public void finish(NetNode node,MessagePackage response) {
	// Command c = null;
	// c = runningCommand.poll();
	// wakeup();
	// c.getCallBack().response(node,response);
	// }

	public boolean addCommand(Command c, int timeout, TimeUnit timeUnit)
			throws InterruptedException {
		if (timeout == 0) {
			commandQueue.put(c);
			wakeup();
			return true;
		} else if (commandQueue.offer(c, timeout, timeUnit)) {
			// logger.debug("add command");
			wakeup();
			return true;
		} else {
			return false;
		}
	}

	public NetNode getNode() {
		return node;
	}

	private void reConnect() throws InterruptedException {
		try {
			NetNode node = null;
			while (node == null) {
				String[] senator = this.senator;
				HashMap<String, NetNode> activeSenators = activeNode;
				node = activeSenators.get(senator[0]);
				if (node != null) {
					logger.info("connect to " + senator[0]);
					// close other senator
					for (java.util.Map.Entry<String, NetNode> e : activeSenators.entrySet()) {
						if (!e.getKey().equals(senator[0])) {
							e.getValue().close();
						}
					}
				} else {
					if (senator.length > 1) {
						for (int i = 1; i < senator.length; ++i) {
							node = activeNode.get(senator[i]);
							if (node == null) {
								String[] tmp = senator[i].split(":");
								ConnectConfig connectConfig = new ConnectConfig(
										senator[i], false, isPromiseMaster,
										connectEvent);
								logger.debug("connecting to " + senator[i]);
								try{
									boolean isSuccess = nioSocketClient.connect(tmp[0], Integer.parseInt(tmp[1]), connectConfig).get(3, TimeUnit.SECONDS);
									if (isSuccess) {
										connectConfig.waitOnOpenEvent();
										logger.info("connect to " + senator[i]);
									}else{
										logger.info("fail to connect to " + senator[i]);
									}
								}catch(TimeoutException e){
									logger.info("connect to " + senator[i] + " time out~");
								}
								
							}
							node = activeNode.get(senator[i]);
							if (node != null) {
								break;
							}
						}
					}
				}

				if (node == null) {
					Thread.sleep(2000);
					logger.error("find master position failed!");
				} else {
					try {
						nodeChangeLock.lock();
						if (senator == this.senator) {
							logger.info("change to node");
							this.node = node;
						} else {
							node.close();
							node = null;
						}
					} finally {
						nodeChangeLock.unlock();
					}
				}
			}
		} catch (ExecutionException e) {
			logger.error("fatal", e);
		} catch (Throwable e) {
			logger.error("fatal", e);
		}
	}
	public Entry getRoot(){
		return root;
	}
	
	public AtomicInteger getRunningSize() {
		return runningSize;
	}

	public static ClientHandler getHandler() {
		return handler;
	}
	
}
